package com.jscloud.bigdata.flink.flinksql.functions.udtf;

import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.json.JSONArray;
import org.json.JSONObject;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

/**
 * # FlinkSQL自定义函数一进多出实战UDTF
 *
 * 我们也可以自定义函数，实现一条数据进入之后，产出多条数据，类似于爆炸函数一样的作用
 * Table functions(表函数)⼀进多出(炸裂)，继承TableFunction，提供⽆返回值的eval⽅法，使⽤collect来输出。
 * Table functions的返回值是⼀个表，需要跟原来的表join才能得到最终结果，因此要⽤到侧写表(不明⽩的可
 * 以研究下LATERAL TABLE)
 *
 * 现有json数据内容如下：其中userBaseList是一个array数组，里面存放了多个用户信息，使用UDTF自定义函数，将每个用户信息给解析出来
 * {"date_time":1665145907806,"price":258.7,"productId":920956185,"userBaseList":[{"begin_time":"2022-10-07 08:38:31","email":"njvjeuchpe@mk1t0.d4e","id":"0","name":"尹修彻"},{"begin_time":"2022-10-07 08:33:59","email":"qurkb119uo@fvyg5.kqj","id":"1","name":"萧幅括"},{"begin_time":"2022-10-07 08:37:40","email":"i4w8ecponz@bpoay.3yv","id":"2","name":"胡乘"},{"begin_time":"2022-10-07 08:38:05","email":"uwl7fpfwbb@b7riu.fh3","id":"3","name":"黄煎"},{"begin_time":"2022-10-07 08:37:12","email":"bjjqrvajih@c75ur.lhs","id":"4","name":"袁肇"}]}
 * {"date_time":1665145918652,"price":258.7,"productId":-786075263,"userBaseList":[{"begin_time":"2022-10-07 08:39:47","email":"yfynwlektk@sz0me.hys","id":"0","name":"程痢"},{"begin_time":"2022-10-07 08:32:03","email":"jwpwuiwdnc@esxbd.hta","id":"1","name":"程盐殃"},{"begin_time":"2022-10-07 08:40:17","email":"fbfnidktqg@zaxxw.g1w","id":"2","name":"蔡锻"},{"begin_time":"2022-10-07 08:35:24","email":"twrm30opcb@5rgzj.sow","id":"3","name":"李猎甩"},{"begin_time":"2022-10-07 08:33:05","email":"rnkabnvaz9@bt319.xlk","id":"4","name":"夏焙匈"}]}
 * {"date_time":1665145927285,"price":258.7,"productId":-988723330,"userBaseList":[{"begin_time":"2022-10-07 08:37:04","email":"pcs8ejgibk@kxf95.djq","id":"0","name":"郝疯框"},{"begin_time":"2022-10-07 08:40:20","email":"n63k5twind@eddbg.aui","id":"1","name":"万侨"},{"begin_time":"2022-10-07 08:33:52","email":"1xmk0vh3bb@1htg2.tw2","id":"2","name":"侯临迸"},{"begin_time":"2022-10-07 08:33:05","email":"cnrqk4crpy@svhkq.wwf","id":"3","name":"闾丘耘"},{"begin_time":"2022-10-07 08:34:26","email":"ubozcxmrxc@c6qpp.8ug","id":"4","name":"皇甫坡"}]}
 * ```
 */
public class FlinkSQLTableFunction {

        public static void main(String[] args) {
                Logger.getLogger("org").setLevel(Level.ERROR);
                //1、创建TableEnvironment
                EnvironmentSettings settings = EnvironmentSettings
                        .newInstance()
                        //.useBlinkPlanner()//Flink1.14开始就删除了其他的执行器了，只保留了BlinkPlanner，默认就是
                        //.inStreamingMode()//默认就是StreamingMode
                        .inBatchMode()
                        .build();

                TableEnvironment tableEnvironment = TableEnvironment.create(settings);

                tableEnvironment.createTemporarySystemFunction("JsonFunc",JsonFunction.class);
                tableEnvironment.createTemporarySystemFunction("explodeFunc",ExplodeFunc.class);


                String source_sql = "CREATE TABLE json_table (\n" +
                        "  line STRING \n" +
                        ") WITH (\n" +
                        "  'connector'='filesystem',\n" +
                        "  'path'='datas/product_user.json',\n" +
                        "  'format'='raw'\n" +
                        ")";

                tableEnvironment.executeSql(source_sql);

                //方式一：使用TableAPI通过内连接来实现
                tableEnvironment.from("json_table")
                        .joinLateral(call(ExplodeFunc.class,$("line"),"userBaseList")
                                .as("id","name","begin_time","email"))
                        .select(call(JsonFunction.class,$("line"),"date_time"),
                                call(JsonFunction.class,$("line"),"price"),
                                call(JsonFunction.class,$("line"),"productId"),
                                $("id"),
                                $("name"),
                                $("begin_time"),
                                $("email")
                        ).execute().print();

                //方式二：使用TableAPI通过左外连接来实现
                tableEnvironment.from("json_table")
                        .leftOuterJoinLateral(call(ExplodeFunc.class,$("line"),"userBaseList")
                                .as("id","name","begin_time","email"))
                        .select(call(JsonFunction.class,$("line"),"date_time"),
                                call(JsonFunction.class,$("line"),"price"),
                                call(JsonFunction.class,$("line"),"productId"),
                                $("id"),
                                $("name"),
                                $("begin_time"),
                                $("email")
                        ).execute().print();

                //方式三：使用FlinkSQL内连接来实现
                tableEnvironment.sqlQuery("select " +
                        "JsonFunc(line,'date_time')," +
                        "JsonFunc(line,'price')," +
                        "JsonFunc(line,'productId')," +
                        "id," +
                        "name," +
                        "begin_time " +
                        "email " +
                        "  from json_table " +
                        ",lateral table(explodeFunc(line,'userBaseList')) "
                ).execute().print();

                //方式四：使用FlinkSQL左外连接来实现
                tableEnvironment.sqlQuery("select " +
                        "JsonFunc(line,'date_time') as date_time," +
                        "JsonFunc(line,'price') as price ," +
                        "JsonFunc(line,'productId') as productId," +
                        "id," +
                        "name," +
                        "begin_time " +
                        "email " +
                        "  from json_table  left join lateral table (explodeFunc(line,'userBaseList')) as sc(id,name,begin_time,email) on true"
                ).execute().print();

        }

        /**
         * 自定义udf
         */
        public static class JsonFunction extends ScalarFunction {
                public String eval(String line,String key){
                        //转换为JSON
                        JSONObject baseJson = new JSONObject(line);
                        String value = "";
                        if(baseJson.has(key)){
                                //根据key获取value
                                return baseJson.getString(key);
                        }
                        return value;
                }
        }


        /**
         * 自定义UDTF，一定要继承 TableFunction 类
         *
         * @FunctionHint
         */
        @FunctionHint(output = @DataTypeHint("ROW<id String,name String,begin_time String,email String>"))
        public static class ExplodeFunc  extends TableFunction {

                public void eval(String line,String key){
                        JSONObject jsonObject = new JSONObject(line);
                        JSONArray jsonArray = new JSONArray(jsonObject.getString(key));
                        for(int i = 0;i< jsonArray.length();i++){
                                String date_time = jsonArray.getJSONObject(i).getString("begin_time");
                                String email = jsonArray.getJSONObject(i).getString("email");
                                String id = jsonArray.getJSONObject(i).getString("id");
                                String name = jsonArray.getJSONObject(i).getString("name");
                                collect(Row.of(id,name,date_time,email));
                        }
                }
        }
}